package cdrprocessing.bolt.filter;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import cdrprocessing.bolt.util.WindowAwareBolt;

public class BloomFilterBolt extends WindowAwareBolt {

    private OutputCollector collector;

    private double falsePositiveProbability;

    private int expectedNumberOfElements;

    private BloomFilter<String> bloomFilter;

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        falsePositiveProbability = Double.valueOf((String) stormConf.get("falsePositiveProbability"));
        expectedNumberOfElements = Integer.valueOf((String) stormConf.get("expectedNumberOfElements"));

        this.collector = collector;
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("RawCdrStream", new Fields("rawCdr"));
    }

    protected void handleStartWindow() {
        bloomFilter = new BloomFilter<String>(falsePositiveProbability, expectedNumberOfElements);
    }

    protected void handleEndWindow() {
        bloomFilter = null;
    }

    protected void handleTuple(Tuple input) {
        final String rawCdr = input.getStringByField("rawCdr");

        if (bloomFilter.getFalsePositiveProbability() < 0.1) {
            if (!bloomFilter.contains(rawCdr)) {
                bloomFilter.add(rawCdr);
                collector.emit("RawCdrStream", new Values(rawCdr));
            } else {
                System.out.println(rawCdr);
            }
        } else {
            System.out.println("shit");
        }
    }
}
